package com.rocketmq.consumer.handle;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.rocketmq.consumer.exception.NotIgnoreDataStoreException;
import com.rocketmq.consumer.exception.ServiceException;
import com.rocketmq.consumer.utils.ThreadPollUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ExecutorService;

/**
 * Topic监听抽象类
 *
 * @Author xuziwen
 * @Date 2017-08-21 14:24
 */
@Slf4j
public abstract class AbstractListenerHandler implements MessageListenerOrderly {

    protected static ExecutorService service = null;
    protected String topicName;
    protected String consumerName;
    protected String tagsName = "*";
    protected Boolean isThreadPoll = true;
    protected Boolean messageBroadcast = false;

    public abstract void handle(String body) throws Exception;

    public void handle(String body, String msgId) {

    }

    private void execute(List<MessageExt> list) {
        for (MessageExt ext : list) {
            ExecutorService service = ThreadPollUtil.getExecutorService();
            service.execute(new Runnable() {
                @Override
                public void run() {
                    String body = null;
                    try {
                        body = new String(ext.getBody(), "UTF-8");
                        String msgId = ext.getMsgId();
                        handle(body);
                        handle(body, msgId);
                    } catch (Exception e) {
                        log.error("deal with " + body + " falure:", e);
                    }
                }
            });
        }
    }

    private void singleExecute(List<MessageExt> list) {
        for (MessageExt ext : list) {
            String body = null;
            try {
                body = new String(ext.getBody(), "UTF-8");
                handle(body);
                String msgId = ext.getMsgId();
                handle(body, msgId);
            } catch (Exception e) {
                log.error("deal with " + body + " falure:", e);
            }
        }
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        consumeOrderlyContext.setAutoCommit(true);
        if (isThreadPoll) {
            this.execute(list);
        } else {
            singleExecute(list);
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }

    protected void exceptionHandle(String body, ServiceException se) {
        if (se instanceof NotIgnoreDataStoreException) {
//            Log log = buildLog(body);
        } else {
            processBizException(body, se);
        }
    }

    protected void processBizException(String body, ServiceException se) {
        log.error("deal with " + body + " falure:", se);
    }


    public String getTopicName() {
        return topicName;
    }

    public String getConsumerName() {
        return consumerName;
    }

    public String getTagsName() {
        return tagsName;
    }

    /**
     * 设置订阅Topic名称
     *
     * @param: []
     * @return: java.lang.String
     * @author: xuziwen
     * @Date: 2017-08-27 13:52
     */
    public abstract void setTopicName(String topicName);

    /**
     * 设置消费者组名称
     *
     * @param: []
     * @return: java.lang.String
     * @author: xuziwen
     * @Date: 2017-08-27 13:52
     */
    public abstract void setConsumerName(String consumerName);

    /**
     * 设置Topic中Tags名称(默认全部)
     *
     * @param: [tagsName]
     * @return: void
     * @author: xuziwen
     * @Date: 2017-08-27 14:23
     */
    public abstract void setTagsName(String tagsName);

    public boolean getMessageBroadcast(){
        return messageBroadcast;
    }
}
